package com.example.demo_mysql_mybatisplus.shyiko;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.json.JSONConfig;
import cn.hutool.json.JSONUtil;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;

import javax.sql.DataSource;
import java.io.IOException;
import java.io.Serializable;
import java.sql.Connection;
import java.sql.DatabaseMetaData;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/*
        <dependency>
            <groupId>com.github.shyiko</groupId>
            <artifactId>mysql-binlog-connector-java</artifactId>
            <version>0.21.0</version>
        </dependency>
*/
//@Component
public class Test_2_shyiko implements InitializingBean { // 使用Spring Bean的生命周期函数afterPropertiesSet()：做后置动作

    @Value("${db.host}")
    private String host ;
    @Value("${db.port:3306}")
    private int port ;
    @Value("${db.user}")
    private String userName ;
    @Value("${db.pwd}")
    private String pwd ;
    @Autowired
    private DataSource dataSource;

    // 表的 id 与 名称 映射
    private static Map<Long ,String> tableId2Name = new HashMap<>();
    // 表的 名称 与 列 映射
    private static Map<String, List<String>> tableName2Columns = new HashMap<>();

    // 获取数据库表的列
    public Map<String, List<String>> getAllTabColumns() {
        Map<String, List<String>> cols = null ;
        try {
            cols = new HashMap<String, List<String>>();
            Connection conn  = dataSource.getConnection() ;

            DatabaseMetaData metaData = conn.getMetaData();
            ResultSet r = metaData.getCatalogs();
            String tableType[] = { "TABLE" };
            while (r.next()) {
                String databaseName = r.getString("TABLE_CAT");
                ResultSet result = metaData.getTables(databaseName, null, null, tableType);
                while (result.next()) {
                    String tableName = result.getString("TABLE_NAME");
                    String key = databaseName + "." + tableName;
                    ResultSet colSet = metaData.getColumns(databaseName, null, tableName, null);
                    cols.put(key, new ArrayList<String>());
                    while (colSet.next()) {
                        String column = colSet.getString("COLUMN_NAME");
                        cols.get(key).add(column);
                    }
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return cols;
    }

    @Override
    public void afterPropertiesSet() {
        // 获取数据库所有表的列
        if(CollectionUtil.isEmpty(tableName2Columns)){
            tableName2Columns = getAllTabColumns();
        }
        // 启动监听
        ExecutorService exec = Executors.newSingleThreadExecutor() ;
        exec.submit(()->{
            try {
                BinaryLogClient client = new BinaryLogClient(host, port , userName, pwd);
                client.setServerId(System.currentTimeMillis());
                client.registerEventListener(event -> {
                    EventListen( event) ;
                });
                client.connect();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }) ;
    }

    private void EventListen( Event event){
        EventData data = event.getData();
        if (data instanceof TableMapEventData) {
            TableMapEventData eventData = (TableMapEventData) data;
            // 存储表的id 与 名称映射 ，后面会用到
            if(!tableId2Name.containsKey(eventData.getTableId())){
                tableId2Name.put(eventData.getTableId(),eventData.getDatabase()+"."+eventData.getTable()) ;
            }
        }else if (data instanceof UpdateRowsEventData) { // 更新事件监听
            UpdateRowsEventData eventData = (UpdateRowsEventData) data;
            dealUpdate( eventData.getTableId() , eventData.getRows() , "update");
        } else if (data instanceof WriteRowsEventData) { // 插入事件监听
            WriteRowsEventData eventData = (WriteRowsEventData) data;
            dealInsertOrDelete( eventData.getTableId() , eventData.getRows() , "insert");
        } else if (data instanceof DeleteRowsEventData) { // 删除事件监听
            DeleteRowsEventData eventData = (DeleteRowsEventData) data;
            dealInsertOrDelete( eventData.getTableId() , eventData.getRows() , "delete");
        }
    }

    /** 处理插入、删除事件 */
    private static void dealInsertOrDelete(Long tableId , List<Serializable[]> rows , String type){
        String tableName = tableId2Name.get( tableId );
        List<String> columns = tableName2Columns.get(tableName);

        List<Result> resultList = new ArrayList<>();
        for( Serializable[] row : rows){
            Map<String ,Object > valueMap = new HashMap<>();
            List<String > changedColumns = new LinkedList<>();

            for(int i=0;i<row.length;i++){
                Serializable value = row[i];
                valueMap.put( columns.get(i) , ifNull(value));
                changedColumns.add( columns.get(i) );
            }

            Result res = new Result().setType(type).setTableName(tableName).setChangedColumns(changedColumns) ;
            if("insert".equalsIgnoreCase(type)){
                res.setAfter(valueMap);
            }else if("delete".equalsIgnoreCase(type)){
                res.setBefore(valueMap);
            }
            resultList.add(res);
        }

        System.out.println( "bingo监听--------"+jsonStr( resultList) );

    }

    /** 处理更新事件 */
    private static void dealUpdate(Long tableId , List<Map.Entry<Serializable[], Serializable[]>> rows , String type){
        String tableName = tableId2Name.get( tableId );
        List<String> columns = tableName2Columns.get(tableName);

        List<Result> resultList = new ArrayList<>();
        for( Map.Entry<Serializable[], Serializable[]> row : rows ){
            Serializable[] before = row.getKey();
            Serializable[] after = row.getValue();
            List<String > changedColumns = new LinkedList<>();
            Map<String ,Object > beforeMap = new HashMap<>();
            Map<String ,Object > afterMap = new HashMap<>();

            for(int i=0;i<before.length;i++){
                boolean ifChange = ( null!=before[i] && null==after[i] )
                        || ( null==before[i] && null!=after[i] )
                        || ( null!=before[i] &&null!=after[i] && !before[i].toString().equals(after[i].toString()) ) ;
                if( ifChange ){
                    beforeMap.put( columns.get(i) , ifNull(before[i]));
                    afterMap.put( columns.get(i) , ifNull(after[i]));
                    changedColumns.add( columns.get(i) );
                }
            }

            Result res = new Result().setType(type).setTableName(tableName).setChangedColumns(changedColumns)
                    .setBefore(beforeMap).setAfter(afterMap) ;
            resultList.add(res);
        }

        System.out.println( jsonStr( resultList) );

        resultList = JSONUtil.toList(jsonStr( resultList) , Result.class ) ;
        System.out.println(  "bingo监听--------"+resultList );

    }

    /** 结果集转换 */
    private static String jsonStr( Object obj){
        if(obj==null){
            return null ;
        }
        return JSONUtil.toJsonStr( obj , new JSONConfig().setDateFormat("yyyy-MM-dd HH:mm:ss")) ;
    }
    /** 空值默认 */
    private static Object ifNull(Object val){
        return null==val?"null":val ;
    }






}
